
package org.openeai.lite.common.control;

import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collection;

import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.management.ObjectName;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.broker.jmx.QueueViewMBean;
import org.apache.activemq.broker.jmx.TopicViewMBean;
import org.apache.log4j.Logger;
import org.openeai.lite.broker.control.ActiveMQController;
import org.openeai.lite.common.exception.OpenEAIException;
import org.openeai.lite.common.model.Queue;
import org.openeai.lite.common.model.Topic;
import org.openeai.lite.common.util.Constants;
import org.openeai.lite.common.util.DateTimeUtils;


/* REVIEW: Could be made ActiveMQ independent by only using JMS api and
 * implementing specific calls in ActiveMQController.
 * [RPR] Won't fix for now.
 */
public class BrokerController
{

    private static Logger log = Logger.getLogger(BrokerController.class);

    private String configuration = Constants.MQ_CONFIG_DEFAULT;

    private String startdate;

    // Reuse certain JMS objects as much as possible for best performance
    private ActiveMQConnection defaultConnection = null;

    private Session defaultSession = null;

    // run controller as singleton
    private static volatile BrokerController instance;

    private static final Object instanceLock = new Object();

    private BrokerController()
    {
    };

    /**
     * Get the {@link BrokerController} singleton instance in a thread safe way
     * (Java 1.5+ !!!)
     * 
     * @return The {@link BrokerController} singleton instance.
     */
    public static BrokerController getInstance()
    {
        if (instance == null)
        {
            synchronized (instanceLock)
            {
                if (instance == null)
                {
                    log.info("Creating new MQ Controller instance");
                    instance = new BrokerController();
                }
            }
        }
        return instance;
    }

    public synchronized void start(String startupType, String configuration) throws OpenEAIException
    {
        try
        {
            if (!isStarted())
            {
            	log.debug("Startup type: " +startupType);
                if (startupType == null || startupType.equals(Constants.STARTUP_VARIANT))
                {
                    // Start ActiveMQ message broker using pre-configured configuration
                    ActiveMQController.getInstance().start(Constants.STARTUP_VARIANT, getConfiguration());
                }
                else if (startupType.equals(Constants.STARTUP_CUSTOM) && configuration != null)
                {
                    // Start ActiveMQ message broker with custom configuration file
                    ActiveMQController.getInstance().start(Constants.STARTUP_CUSTOM, configuration);
                }
                else
                {
                    throw new OpenEAIException("Error starting Message Broker with startup type " + startupType + " and optional configuration " + configuration);
                }
                setStartdate();
                log.info("Broker started at " + getStartdate());
            }
            else
            {
                log.info("Message router already started.");
            }

        }
        catch (OpenEAIException ee) {
        	log.error(ee);
        	throw new OpenEAIException(ee);
        }
        catch (Exception e)
        {
            log.error("Error starting Message Broker: " + e.getMessage());
            throw new OpenEAIException("Error starting Message Broker: ", e);
        }

    }

    // Stop ActiveMQ message broker
    public synchronized void stop() throws OpenEAIException
    {
        log.info("Stopping Message Broker.");
        try
        {
            if (defaultSession != null)
            {
                defaultSession.close();
                defaultSession = null;
            }
            if (defaultConnection != null)
            {
                defaultConnection.close();
                defaultConnection = null;
            }
        }
        catch (JMSException e)
        {
            log.error("Error closing default session/connection normally.");
        }

        ActiveMQController.getInstance().stop();
    }

    // Check if message broker is started
    public Boolean isStarted() throws OpenEAIException
    {
        try
        {
            return ActiveMQController.getInstance().isStarted();
        }
        catch (Exception e)
        {
            log.error("Could not determing message broker status!");
            throw new OpenEAIException(e);
        }
    }

    public String getStartdate()
    {
        return startdate;
    }

    private void setStartdate()
    {
        startdate = new DateTimeUtils().currentDateString();
    }

    // get a connection to default message broker
    public ActiveMQConnection getConnection() throws OpenEAIException
    {
        log.info("Getting connection to default broker");
        try
        {
            if (defaultConnection == null)
            {
                log.info("Creating connection for url " + Constants.MQ_BROKER_URL);
                defaultConnection = ActiveMQConnection.makeConnection(Constants.MQ_BROKER_URL);

                log.info("Starting connection.");
                defaultConnection.start();
            }
            return defaultConnection;
        }
        catch (JMSException e)
        {
            log.error("Error creating connection to message broker!");
            throw new OpenEAIException(e);
        }
        catch (URISyntaxException e)
        {
            log.error("Error creating connection to message broker!");
            throw new OpenEAIException(e);
        }
    }

    /**
     * Create a new connection to a given broker URL
     * 
     * @param brokerurl
     *            - the URL of the message broker to connect to
     * @return ActiveMQConnection
     * @throws OpenEAIException
     */
    public ActiveMQConnection createConnection(String brokerurl) throws OpenEAIException
    {
        log.info("Creating new connection to broker" + brokerurl);
        ActiveMQConnection connection = null;
        try
        {
            connection = ActiveMQConnection.makeConnection(brokerurl);
            connection.start();
            return connection;
        }
        catch (JMSException e)
        {
            log.error("Error creating connection to message broker with url " + brokerurl);
            throw new OpenEAIException("Error creating connection to message broker with url "
                            + brokerurl, e);
        }
        catch (URISyntaxException e)
        {
            log.error("Error creating connection to message broker with url " + brokerurl);
            throw new OpenEAIException("Error creating connection to message broker with url "
                            + brokerurl, e);
        }
    }

    // Create a new queue
    public void createQueue(String queueName) throws OpenEAIException
    {
        log.info("Creating queue " + queueName);
        try
        {
            ActiveMQController.getInstance().getBroker().getAdminView().addQueue(queueName);
            log.info("Queue " + queueName + " created.");
        }
        catch (Exception e)
        {
            log.error("Error creating broker queue " + queueName);
            throw new OpenEAIException(e);
        }
    }

    // Delete a queue
    public void deleteQueue(Queue queue) throws OpenEAIException
    {
        String queueName = queue.getDestination();
        log.info("Deleting queue " + queueName);
        try
        {
            ActiveMQController.getInstance().getBroker().getAdminView().removeQueue(queueName);
            log.info("Queue " + queueName + " deleted.");
        }
        catch (Exception e)
        {
            log.error("Error deleting broker queue " + queueName);
            throw new OpenEAIException(e);
        }
    }

    // Create a new topic
    public void createTopic(String topicName) throws OpenEAIException
    {
        log.info("Creating topic " + topicName);
        try
        {
            ActiveMQController.getInstance().getBroker().getAdminView().addTopic(topicName);
            log.info("topic " + topicName + " created.");
        }
        catch (Exception e)
        {
            log.error("Error creating broker topic " + topicName);
            throw new OpenEAIException(e);
        }
    }

    // Delete a topic
    public void deleteTopic(Topic topic) throws OpenEAIException
    {
        String topicName = topic.getDestination();
        log.info("Deleting topic " + topicName);
        try
        {
            ActiveMQController.getInstance().getBroker().getAdminView().removeTopic(topicName);
            log.info("Topic " + topicName + " deleted.");
        }
        catch (Exception e)
        {
            log.error("Error deleting broker topic " + topicName);
            throw new OpenEAIException(e);
        }
    }

    // Get memory usage information
    public Long getMemoryUsage() throws OpenEAIException
    {
        Long usage = new Long(0);
        try
        {
            usage = ActiveMQController.getInstance().getBroker().getConsumerSystemUsage()
                            .getMemoryUsage().getUsage();
        }
        catch (Exception e)
        {
            log.error("Error getting memory usage");
            throw new OpenEAIException(e);
        }
        return usage;
    }

    // Not exactly useful, but nice to include for completeness
    public void stopConnection(ActiveMQConnection connection) throws OpenEAIException
    {
        try
        {
            log.info("Stopping connection " + connection.getClientID());
            connection.stop();
        }
        catch (JMSException e)
        {
            log.error("Error stopping connection to message broker!");
            throw new OpenEAIException(e);
        }
    }

    // get a session on the default ActiveMQ connection
    public Session getSession() throws OpenEAIException
    {
        log.info("Getting session on the default connection");
        try
        {
            if (defaultSession == null)
            {
                log.info("Creating new session on default connection");
                defaultSession = getConnection().createSession(false, Session.AUTO_ACKNOWLEDGE);
            }
            return defaultSession;
        }
        catch (Exception e)
        {
            log.error("Error creating connection to message broker!");
            throw new OpenEAIException(e);
        }
    }

    // Create a new session on a session
    public Session createSession(ActiveMQConnection connection) throws OpenEAIException
    {
        Session session = null;
        try
        {
            log.info("Creating new session using connection " + connection.getClientID());
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        }
        catch (JMSException e)
        {
            log.error("Error creating session to message broker!");
            throw new OpenEAIException(e);
        }
        return session;
    }

    // Create a new transacted session on a session
    public Session createTransactedSession(ActiveMQConnection connection) throws OpenEAIException
    {
        Session session = null;
        try
        {
            log.info("Creating new transacted session using connection " + connection.getClientID());
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        }
        catch (JMSException e)
        {
            log.error("Error creating session to message broker!");
            throw new OpenEAIException(e);
        }
        return session;
    }

    // Not exactly useful, but nice to include for completeness
    public void closeSession(Session session) throws OpenEAIException
    {
        try
        {
            log.info("Closing JMS session");
            session.close();
        }
        catch (JMSException e)
        {
            log.error("Error closing session to message broker!");
            throw new OpenEAIException(e);
        }
    }

    // Get all queues
    public ArrayList<Queue> getQueues() throws OpenEAIException
    {
        ArrayList<Queue> queueList = new ArrayList<Queue>();
        if (isStarted())
        {
            try
            {
                ObjectName[] queueCollection = ActiveMQController.getInstance().getBroker()
                                .getAdminView().getQueues();
                for (ObjectName o : queueCollection)
                {
                    Queue queue = new Queue();
                    String brokerName = Constants.UNKNOWN;
                    String destination = Constants.UNKNOWN;
                    Integer msgs = new Integer(0);

                    brokerName = o.getKeyProperty("BrokerName");
                    destination = o.getKeyProperty("Destination");

                    msgs = getMessageCountForQueue(destination);

                    queue.setBrokerName(brokerName);
                    queue.setDestination(destination);
                    queue.setMsgs(msgs);

                    queueList.add(queue);
                    queue = null;
                }

            }
            catch (Exception e)
            {
                log.error("Error getting information about queues!");
                throw new OpenEAIException(e);
            }
        }
        else
        {
            log.info("Message broker is not started!");
        }

        return queueList;
    }

    // Get the number of active queues
    public Integer getQueueCount() throws OpenEAIException
    {
        Integer queues = 0;
        if (isStarted())
        {
            try
            {
                queues = ActiveMQController.getInstance().getBrokerAdmin().getQueues().length;
            }
            catch (Exception e)
            {
                log.error("Error getting information about queues!");
                throw new OpenEAIException(e);
            }
        }
        else
        {
            log.info("Message broker is not started!");
        }
        return queues;
    }

    // Get all topics
    public ArrayList<Topic> getTopics() throws OpenEAIException
    {
        ArrayList<Topic> topicList = new ArrayList<Topic>();
        if (isStarted())
        {
            try
            {
                ObjectName[] topicCollection = ActiveMQController.getInstance().getBroker()
                                .getAdminView().getTopics();

                for (ObjectName o : topicCollection)
                {
                    Topic topic = new Topic();
                    String brokerName = Constants.UNKNOWN;
                    String destination = Constants.UNKNOWN;
                    Integer msgs = new Integer(0);

                    brokerName = o.getKeyProperty("BrokerName");
                    destination = o.getKeyProperty("Destination");

                    msgs = getMessageCountForTopic(destination);

                    topic.setBrokerName(brokerName);
                    topic.setDestination(destination);
                    topic.setMsgs(msgs);

                    topicList.add(topic);
                    topic = null;
                }
            }
            catch (Exception e)
            {
                log.error("Error getting information about topics!");
                throw new OpenEAIException(e);
            }
        }
        else
        {
            log.info("Cannot get information: message broker is not started");
        }

        return topicList;
    }

    // Get the number of active topics
    public Integer getTopicCount() throws OpenEAIException
    {
        Integer topics = 0;
        if (isStarted())
        {
            try
            {
                topics = ActiveMQController.getInstance().getBrokerAdmin().getTopics().length;
            }
            catch (Exception e)
            {
                log.error("Error getting information about topics!");
                throw new OpenEAIException(e);
            }
        }
        else
        {
            // ActiveMQ not started!
            log.info("Cannot get information: message broker is not started");
        }
        return topics;

    }

    // Get the number of temporary queues
    public Integer getTemporaryQueueCount() throws OpenEAIException
    {
        Integer queues = 0;
        if (isStarted())
        {
            try
            {
                queues = ActiveMQController.getInstance().getBroker().getAdminView()
                                .getTemporaryQueues().length;
            }
            catch (Exception e)
            {
                log.error("Error getting information about queues!");
                throw new OpenEAIException(e);
            }
        }
        else
        {
            log.info("Message broker is not started!");
        }
        return queues;
    }

    // Get the configured OpenWire URL
    public String getOpenWireURL() throws OpenEAIException
    {
        try
        {
            return ActiveMQController.getInstance().getBroker().getAdminView().getOpenWireURL();
        }
        catch (Exception e)
        {
            log.error("Error getting OpenWire URL information!");
            throw new OpenEAIException(e);
        }
    }

    // Get the number of temporary topics
    public Integer getTemproraryTopicCount() throws OpenEAIException
    {
        Integer topics = 0;
        if (isStarted())
        {
            try
            {
                topics = ActiveMQController.getInstance().getBroker().getAdminView()
                                .getTemporaryTopics().length;
            }
            catch (Exception e)
            {
                log.error("Error getting information about topics!");
                throw new OpenEAIException(e);
            }
        }
        else
        {
            // ActiveMQ not started!
            log.info("Cannot get information: message broker is not started");
        }
        return topics;

    }

    // Return the configured Broker name
    public String getBrokerName() throws OpenEAIException
    {
        try
        {
            return ActiveMQController.getInstance().getBrokerAdmin().getBrokerName();
        }
        catch (Exception e)
        {
            log.error("Error getting message broker information!");
            throw new OpenEAIException(e);
        }
    }

    // Returns the specified configuration file
    public String getConfiguration() throws OpenEAIException
    {
        return configuration;
    }

    public void setConfiguration(String aConfiguration)
    {
        configuration = aConfiguration;
    }

    // Return the generated Broker ID
    public String getBrokerId() throws OpenEAIException
    {
        try
        {
            return ActiveMQController.getInstance().getBrokerAdmin().getBrokerId();
        }
        catch (Exception e)
        {
            log.error("Error getting message broker information!");
            throw new OpenEAIException(e);
        }
    }

    public void sendMessageToQueue(String queueName, String message) throws OpenEAIException
    {
        try
        {
            MessageProducer producer = getSession().createProducer(getSession().createQueue(queueName));
            TextMessage tmessage = getSession().createTextMessage(message);

            log.info("Sending message to queue " + queueName);
            producer.send(tmessage);
            producer.close();

        }
        catch (Exception e)
        {
            log.error("Error sending message!");
            throw new OpenEAIException(e);
        }
    }

    public void sendMessageToTopic(String topicName, String message) throws OpenEAIException
    {
        try
        {
            MessageProducer producer = getSession().createProducer(getSession().createTopic(topicName));
            TextMessage tmessage = getSession().createTextMessage(message);

            log.info("Sending message to topic " + topicName);            
            producer.send(tmessage);
            producer.close();
        }
        catch (Exception e)
        {
            log.error("Error sending message!");
            throw new OpenEAIException(e);
        }
    }

    private Integer getMessageCountForQueue(String destination) throws OpenEAIException
    {
        int msgCount = 0;
        Collection<QueueViewMBean> queues = null;
        try
        {
            queues = ActiveMQController.getInstance().getBrokerFacade().getQueues();
            for (QueueViewMBean q : queues)
            {
                if (q.getName() == destination)
                {
                    msgCount = (int) q.getQueueSize();
                }
            }
        }
        catch (Exception e)
        {
            log.error("Error determining message count for queue");
            throw new OpenEAIException(e);
        }

        return msgCount;
    }

    private Integer getMessageCountForTopic(String destination) throws OpenEAIException
    {
        int msgCount = 0;
        Collection<TopicViewMBean> topics = null;
        try
        {
            topics = ActiveMQController.getInstance().getBrokerFacade().getTopics();
            for (TopicViewMBean t : topics)
            {
                if (t.getName().equals(destination))
                {
                    msgCount = (int) t.getQueueSize();
                }

            }
        }
        catch (Exception e)
        {
            log.error("Error determining message count for queue");
            throw new OpenEAIException(e);
        }

        return msgCount;
    }

}
